package sg.edu.nus.iss.stockex;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


/**
 * MovingAverageMapper
 * 
 * Input:
 * from file
 * 
 * Output:
 * Key => CodeDateKey (code, date)
 * Value => TimestampPriceDataPoint (tradeDate, price)
 */

public class StockCompositeMapper extends Mapper<LongWritable, Text, CodeDateKey, TimestampPriceDataPoint> {

	static enum Parse_Counters {
		BAD_PARSE
	};

	private final CodeDateKey key = new CodeDateKey();
	private final TimestampPriceDataPoint val = new TimestampPriceDataPoint();


	@Override
	public void map(LongWritable inkey, Text value, Context context)
	throws IOException, InterruptedException {
	
		String line = value.toString();

		Stock stock = Stock.parseString(line);
		
		if (stock != null) {
			key.set(stock.getCode(), stock.getTradeDate().getTime());
			val.setClosePrice(stock.getClosePrice());
			val.setOpenPrice(stock.getOpenPrice());
			val.setLongTradeDate(stock.getTradeDate().getTime());
			val.setHighPrice(stock.getHighPrice());
			val.setLowPrice(stock.getLowPrice());
			val.setVolume(stock.getVolume());
			context.write(key, val);
		}

	}

}
